package com.dljd.demo;

import com.dljd.constant.MqConstant;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

import java.util.List;


/**
 * @Author limin
 * @date 2025-06-15 20:27
 * @description
 */
public class ASimpleTest {

    /**
     * 生产者   发消息
     */
    @Test
    public void simpleProducer() throws Exception{
            //创建一个生产者(指定一个组名)
            DefaultMQProducer producer = new DefaultMQProducer("test-produce-group");
            //连接namesrv
            producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
            //启动
            producer.start();  //以上为建立连接
            //创建一个消息
            Message message = new Message("testTopic","我是一条消息".getBytes());
            //发送消息
            SendResult sendResult = producer.send(message);
            System.out.println(sendResult.getSendStatus());
            //关闭生产者
            producer.shutdown();
    }

    /**
     * 消费者
     * @throws Exception
     */
    @Test
    public void simpleConsumer() throws Exception{
        //创建一个消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-consumer-group");
        //连接namesrv
        consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
        //订阅一个主题 (subscribe 订阅，申请)  *表示订阅这个主题中的所有信息  后期会有消息过滤
        consumer.subscribe("testTopic", "*");
        //设置一个监听器(一直监听，异步回调方式，这个线程和主线程不是同一个)
        //MessageListenerConcurrently  (concurrent 并发的)
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                //匿名内部类
                //这个就是消费的方法(业务处理)
                System.out.println(" 我是消费者");
                System.out.println(msgs.get(0).toString());
                System.out.println("消息上下文：" + context);
                System.out.println(new String(msgs.get(0).getBody()));
                //返回值
                //CONSUME_SUCCESS 消息会从mq出队
                //RECONSUME_LATER（报错/null）失败 消息会重新回到队列，过一会重新投递出来，给当前消费者或者其他消费者消费
                //return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //启动
        consumer.start();
        //Thread.sleep(10000);
        //挂起当前的jvm
        System.in.read();

    }


}
